package com.bridgeintelligent.tag.bulkload.service.core;

import com.bridgeintelligent.tag.bulkload.api.BulkLoadConfig;
import com.bridgeintelligent.tag.bulkload.service.core.refresh.BulkRefreshApi;
import com.bridgeintelligent.tag.bulkload.service.core.refresh.BulkRefreshApiUtils;
import com.bridgeintelligent.tag.bulkload.service.core.refresh.BulkRefreshConfig;
import com.bridgeintelligent.tag.constants.bulkload.BulkLoadMqMsg;
import com.bridgeintelligent.tag.constants.bulkload.MQConfig;
import com.bridgeintelligent.tag.mapper.BatchDimFreeMapper;
import com.bridgeintelligent.tag.mapper.EsIndexNameMapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * Created by JackCheung on 2019/9/19 6:25 PM.
 */
@AllArgsConstructor
@Slf4j
@Component
@RocketMQMessageListener(topic = MQConfig.REAL_TIME_TAG_TOPIC, consumerGroup = MQConfig.REAL_TIME_TAG_CONSUMER_GROUP,
        consumeThreadMax=1)
public class BulkEventDelegate implements RocketMQListener<String> {

    private ObjectMapper objectMapper;
    private IBulkEventHandler iBulkEventHandler;
    private BatchDimFreeMapper batchDimFreeMapper;
    private BulkRefreshApiUtils bulkRefreshApiUtils;
    private BulkRefreshConfig bulkRefreshConfig;
    private EsIndexNameMapper esIndexNameMapper;

    /**
     * 接受MQ消息
     *
     * @param message
     */
    @Override
    public void onMessage(String message) {
        dispense(message);
    }

    /**
     * 处理接收到的自定义标签消息
     *
     * @param msg
     */
    public void dispense(String msg) {
        long time = System.currentTimeMillis();
        String dimId = null;
        try {
            log.info("自定义标签topic收到消息=======>{}", msg);
            //msg转换为BulkLoadMqMsg
            BulkLoadMqMsg bulkLoadMqMsg = objectMapper.readValue(msg, BulkLoadMqMsg.class);
            dimId = bulkLoadMqMsg.getDimId();
            //设置索引类型 初始化数据为1
            String indexType = esIndexNameMapper.findIndexType("1");
            bulkLoadMqMsg.setIndexType(indexType);
            //处理消息
            log.debug("\n------------->>>>开始处理消息");
            iBulkEventHandler.handler(bulkLoadMqMsg);
            log.debug("\n------------->>>>处理完消息");
            //更新为没有数据在跑批
            batchDimFreeMapper.updateDimDataRunning(bulkLoadMqMsg.getDimId(), false,"1");
        } catch (Exception e) {
            e.printStackTrace();
            batchDimFreeMapper.updateDimDataRunning(dimId, false,"2");
            log.error("自定义标签消费消息时出现异常=======>", e);
        } finally {
            refreshEngineCache();
            log.debug("消费自定义标签数据总耗时=====>{}ms", (System.currentTimeMillis() - time));
        }
    }

    /**
     * 自定义标签处理
     *
     * @param bulkLoadMqMsg
     */
    public void dispense(BulkLoadMqMsg bulkLoadMqMsg) {
        /*long time = System.currentTimeMillis();
        try {*/
        iBulkEventHandler.handler(bulkLoadMqMsg);
        /*} catch (Exception e) {
            e.printStackTrace();
            log.error("自定义标签批量出现异常=======>{}", e);
            throw new BulkLoadException(new ExceptionCode(999, e.getMessage()));
        } finally {
            refreshEngineCache();
            log.debug("自定义批量耗时=====>{}ms", (System.currentTimeMillis() - time));
        }*/
    }

    /**
     * 刷新引擎缓存
     */
    private void refreshEngineCache() {
        String url = bulkRefreshConfig.getRequestUrl();
        BulkRefreshApi bulkRefreshApi = bulkRefreshApiUtils.build(BulkRefreshApi.class, url + BulkLoadConfig.REFRESH_PATH);
        bulkRefreshApi.refreshItem();
    }
}
